#ifndef _M__MESSAGE_H__
#define _M__MESSAGE_H__
#include"../mqcommon/helper.hpp"
#include"../mqcommon/logger.hpp"
#include"../mqcommon/mq_msg.pb.h"
#include<unordered_map>
#include<memory>
#include<mutex>
#include<list>
#include<assert.h>

namespace mq{
    #define DATAFILE_SUBFIX ".mpd"
    #define TMPFILE_SUBFIX ".mpd.tmp"
    using MqMessagePtr = std::shared_ptr<mq::Message>;
    
    class MessageMaper{
        public:
            MessageMaper(std::string& dir,const std::string& qname):_qname(qname){
                if(dir.back() != '/')  dir.push_back('/');
                _datafile = dir + _qname + DATAFILE_SUBFIX;
                _tmpfile = dir + _qname + TMPFILE_SUBFIX;
                assert(Filehelper::createDirectory(dir));
                createMsgFile();
            }
            bool createMsgFile(){
                if(Filehelper::exists(_datafile) == false){
                    int ret = Filehelper::createFile(_datafile);
                    if(ret == false){
                        LOG_ERRORM("createMsgfile false...");
                        return false;
                    }
                }
                return true;
            }
            void removeMsgfile(){
                Filehelper::removeFile(_datafile);
                Filehelper::removeFile(_tmpfile);
            }
            bool insertMsg(MqMessagePtr& msg){
                return insert(_datafile,msg);
            }

            void removeMsg(const MqMessagePtr& msg ){
                Filehelper filehander(_datafile);
                msg->mutable_payload()->set_valid("0");
                std::string body = msg->payload().SerializeAsString();
                if(msg->lenth() != body.size()){
                    LOG_ERRORM("文件palyload被修改,和原文件body长度不一致");
                    return;
                }

                size_t ofset = msg->offset();
                int ret = filehander.write(body.c_str(),ofset+sizeof(size_t),body.size());
                if(ret == false){
                    LOG_ERRORM("removeMsg false...");
                    return;
                }

            }

            std::list<MqMessagePtr> garbc(){
                std::list<MqMessagePtr> listMessages;
                load(&listMessages);
                // if(listMessages.size() == 0){
                //     LOG_DEBUG("garbc load list size == 0");
                // }
                int ret;
                Filehelper filehander(_datafile);

                Filehelper::createFile(_tmpfile);
                for(auto& e:listMessages){
                    ret = insert(_tmpfile,e);
                    if(ret == false)
                    {LOG_ERRORM("insert msg false");}
                }
                ret = filehander.removeFile(_datafile);
                if(ret == false){LOG_ERRORM("remove datafile false");}
                ret = filehander.renameFile(_tmpfile,_datafile);
                if(ret == false){
                    LOG_ERRORM("rename datafile false");
                }
                return listMessages;
            }

        private:
            bool load(std::list<MqMessagePtr>* listMessages){
                // std::cout<<"loading..."<<std::endl;
                Filehelper filehander(_datafile);
                size_t offset = 0;
                size_t msg_size = 0;
                size_t fsize = filehander.size();
                while(offset < fsize){
                    int ret = filehander.read((char*)&msg_size,offset,sizeof(size_t));
                    // LOG_DEBUG("read file size: %ld",msg_size);
                    if(ret == false) {
                        return false;
                        LOG_ERRORM("gc read hand false....");
                    }
                    offset += sizeof(size_t);
                    // std::cout<<"reading body...\n";

                    std::string body;
                    body.resize(msg_size);
                    ret = filehander.read(&body[0],offset,msg_size);
                    if(ret == false) {
                        return false;
                        LOG_ERRORM("gc read body false....");
                    }
                    offset += body.size();

                    MqMessagePtr msg = std::make_shared<Message>();
                    msg->mutable_payload()->ParseFromString(body);
                    // LOG_DEBUG("%s",msg->payload().body().c_str());//testing
                    if(msg->payload().valid() == "0") continue;
                    listMessages->push_back(msg);
                }
                return true;
            }

            bool insert(const std::string& filename,MqMessagePtr& msg){
                std::string body = msg->payload().SerializeAsString();//
                Filehelper filehander(filename);
                size_t fsize = filehander.size();
                //test
                // LOG_DEBUG("before insert size: %ld",fsize);
                size_t lenth = body.size();
                int hand = filehander.write((char*)&lenth,fsize,sizeof(size_t));
                if(hand == false){
                    LOG_ERRORM("hand insertMsg false...");
                    return false;
                }
                //4/body 4/body
                int ret = filehander.write(body.c_str(),fsize + sizeof(size_t),body.size());
                if(ret == false){
                    LOG_ERRORM("payload insertMsg false...");
                    return false;
                }
                msg->set_lenth(body.size());
                msg->set_offset(fsize);
                return true;
            }
        private:
            std::string _qname;
            std::string _datafile;
            std::string _tmpfile;

    };
    
    
    class QueueMessage{
        public:
            using ptr = std::shared_ptr<QueueMessage>;
            QueueMessage(std::string& dir,const std::string& qname)
            : _msgmaper(dir,qname),_qname(qname),_valid_count(0),_total_count(0){
            }

            void recovery(){
                std::unique_lock<std::mutex> lock(_mutex);
                
                _pendingMsglist = _msgmaper.garbc();
                // if(_pendingMsglist.size() == 0){
                //     LOG_DEBUG("recovery size == 0")
                // }
                for(auto& msg : _pendingMsglist){
                    _durable_msgs.insert(std::make_pair(msg->payload().basepro().id(),msg));
                }
                _total_count = _pendingMsglist.size();
                _valid_count = _pendingMsglist.size();
            
            }
            bool insert(BasicProperty* bp,const std::string& body,bool qdur){
                MqMessagePtr msg =std::make_shared<mq::Message>();
                msg->mutable_payload()->set_body(body);

                DeliveryMode mode = qdur? DeliveryMode::DURABLE:DeliveryMode::UNDURABLE;
                if(bp){
                    if(qdur == true)
                        mode = (mode == bp->deliverymod())? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                    msg->mutable_payload()->mutable_basepro()->set_id(bp->id());
                    msg->mutable_payload()->mutable_basepro()->set_deliverymod(mode);
                    msg->mutable_payload()->mutable_basepro()->set_routingkey(bp->routingkey());
                }
                else{
                    msg->mutable_payload()->mutable_basepro()->set_id(Uuidhelper::BuideUuid());
                    msg->mutable_payload()->mutable_basepro()->set_deliverymod(mode);
                    msg->mutable_payload()->mutable_basepro()->set_routingkey("");
                }

                std::unique_lock<std::mutex> lock(_mutex);
                if(msg->payload().basepro().deliverymod() == DeliveryMode::DURABLE){
                    msg->mutable_payload()->set_valid("1");
                    _durable_msgs.insert(std::make_pair(msg->payload().basepro().id(),msg));
                    _msgmaper.insertMsg(msg);
                    _valid_count++;
                    _total_count++;
                }

                _pendingMsglist.push_back(msg);
                return true;
            }
            //need garbc

            bool remove(const std::string& msg_id){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _waitack_msgs.find(msg_id);
                if(it == _waitack_msgs.end()) return true;
                if(it->second->payload().basepro().deliverymod() == DeliveryMode::DURABLE){
                    _msgmaper.removeMsg(it->second);
                    --_valid_count;
                    _durable_msgs.erase(msg_id);
                    if(gccheck()) gc();
                }
                _waitack_msgs.erase(msg_id);
                return true;
            }

            MqMessagePtr front(){
                std::unique_lock<std::mutex> lock(_mutex);
                if(_pendingMsglist.size() == 0) return MqMessagePtr();
                MqMessagePtr msg = _pendingMsglist.front();
                _pendingMsglist.pop_front();

                _waitack_msgs.insert(std::make_pair(msg->payload().basepro().id(),msg));
                return msg;
            }
            //test func..
                
            size_t getableCount(){
                std::unique_lock<std::mutex> lock(_mutex);
                return _pendingMsglist.size();
            }
            size_t durableCount(){
                std::unique_lock<std::mutex> lock(_mutex);
                return _durable_msgs.size();
            }
            size_t totalCount(){
                std::unique_lock<std::mutex> lock(_mutex);
                return _total_count;
            }
            size_t waitackCount(){
                std::unique_lock<std::mutex> lock(_mutex); 
                return _waitack_msgs.size();
            }

            void clear(){
                std::unique_lock<std::mutex> lock(_mutex);
                _pendingMsglist.clear();
                _durable_msgs.clear();
                _waitack_msgs.clear();
                _msgmaper.removeMsgfile();
                _valid_count = 0;
                _total_count = 0;
            }
        private:
            void gc(){
                    std::list<MqMessagePtr> msgs = _msgmaper.garbc();
                    for(auto& msg: msgs){
                        auto it = _durable_msgs.find(msg->payload().basepro().id());
                        if(it == _durable_msgs.end()) {
                            LOG_ERRORM("内存存储没有找到 持久化的信息,有bug");
                            _durable_msgs.insert(std::make_pair(msg->payload().basepro().id(),msg));
                        }
                        it->second->set_offset(msg->offset());
                        it->second->set_lenth(msg->lenth());
                    }
                    _valid_count = msgs.size();
                    _total_count = msgs.size();
            }

            bool gccheck(){
                if(_total_count > 2000 && (_valid_count*10 / _total_count > 5))
                    return true;
                else 
                    return false;
            }
        private:
            std::mutex _mutex;
            std::string _qname;
            //count 是对持续化里面的count，不是内存的count
            size_t _valid_count;
            size_t _total_count;
            MessageMaper _msgmaper;
            std::list<MqMessagePtr> _pendingMsglist;
            std::unordered_map<std::string,MqMessagePtr> _durable_msgs;
            std::unordered_map<std::string,MqMessagePtr> _waitack_msgs;

    };
    
    class MessageManager{
        public:
            using ptr = std::shared_ptr<MessageManager>;
            MessageManager(const std::string& dir):_basedir(dir){
                LOG_DEBUGM("message init...");
            }
            void initQMessage(const std::string& qname){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it != _msgs.end()){
                        return;
                    }
                    msg = std::make_shared<QueueMessage>(_basedir,qname);
                    _msgs.insert(std::make_pair(qname,msg));
                }
                msg->recovery();
            }

            void destroyQMessage(const std::string& qname){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        return;
                    }
                    msg = it->second;
                }
                _msgs.erase(qname);
                msg->clear();
            }
            bool insert(const std::string& qname,BasicProperty* bp,const std::string& body,bool qdur){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        return false;
                    }
                    msg = it->second;
                }
                return msg->insert(bp,body,qdur);
            }
            MqMessagePtr front(const std::string& qname){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        return MqMessagePtr();
                    }
                    msg = it->second;
                }
                return msg->front();
            }
            void ack(const std::string& qname,const std::string& msg_id){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        return ;
                    }
                    msg = it->second;
                }
                msg->remove(msg_id);
            }

            size_t getableCount(const std::string& qname){
                QueueMessage::ptr msg;
                {

                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        std::cout<<"no find "<<qname<<"\n";
                        return 0;
                    }
                    msg = it->second;
                }
                return msg->getableCount();
            }

            size_t durableCount(const std::string& qname){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        return 0;
                    }
                    msg = it->second;
                }
                return msg->durableCount();

            }
            size_t totalCount(const std::string& qname){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        return 0;
                    }
                    msg = it->second;
                }
                return msg->totalCount();


            }
            size_t waitackCount(const std::string& qname){
                QueueMessage::ptr msg;
                {
                    std::unique_lock<std::mutex> lock(_mutex); 
                    auto it = _msgs.find(qname);
                    if(it == _msgs.end()){
                        return 0;
                    }
                    msg = it->second;
                }
                return msg->waitackCount();
            }

            void clear(){
                std::unique_lock<std::mutex> lock(_mutex); 
                for(auto& msg :_msgs){
                    msg.second->clear();
                }
            }
        private:
            std::mutex _mutex;
            std::string _basedir;
            std::unordered_map<std::string,QueueMessage::ptr> _msgs;
    };  
    
}

#endif